package com.hw

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.regexp_replace
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo11MaYiSenLin2 {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .config("spark.sql.shuffle.partitions", 1)
      //开启join
      .config("spark.sql.crossJoin.enabled", true)
      .master("local")
      .appName("mySl")
      .getOrCreate()

    val userDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("user_id string,date_dt string,low_carbon int")
      .load("data/user_low_carbon.txt")

    val plantCarbonDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("plant_id string,plant_name string,plant_carbon int")
      .load("data/plant_carbon.txt")

    /**
     * 蚂蚁森林低碳用户排名分析
     * 问题：查询user_low_carbon表中每日流水记录，条件为：
     * 用户在2017年，连续三天（或以上）的天数里，
     * 每天减少碳排放（low_carbon）都超过100g的用户低碳流水。
     * 需要查询返回满足以上条件的user_low_carbon表中的记录流水。
     * 例如用户u_002符合条件的记录如下，因为2017/1/2~2017/1/5连续四天的碳排放量之和都大于等于100g：
     */
    import org.apache.spark.sql.functions._
    import spark.implicits._

    userDF
      //将数据中/替换成-方便取出
      .select($"user_id", regexp_replace($"date_dt", "/", "-") as "dt", $"low_carbon")
      //过滤时间要求
      .where(year($"dt") === 2017)
      //求出单用户当天的总能量
      //.withColumn("day_carbon",sum($"low_carbon") over Window.partitionBy($"user_id",$"dt"))
      .groupBy($"user_id", $"dt")
      .agg(sum($"low_carbon") as "day_carbon")
      //格式化时间
      .withColumn("dt2", date_format($"dt", "yyyy-MM-DD"))
      //取上一天登录的时间
      .withColumn("last_login", lag($"dt2", 1) over Window.partitionBy($"user_id").orderBy($"dt2"))
      //如果前一天登录过就标记为0如果没有登录过就标记为1
      .withColumn("flag", when($"dt2" === date_add($"last_login", 1), 0).otherwise(1))
      //对标记求和,为后面标记分区
      .withColumn("tem", sum($"flag") over Window.partitionBy($"user_id").orderBy($"dt2"))
      //对用户和标记分区，得到有连续多少天登录
      .withColumn("count", count($"*") over Window.partitionBy($"user_id", $"tem"))
      //对用户和标记分区，得到用户连续登录的时间内总能量
      .withColumn("count_carbon", sum($"day_carbon") over Window.partitionBy($"user_id", $"tem"))
      .withColumn("end_date",max($"dt2") over Window.partitionBy($"user_id",$"tem"))
      .withColumn("start_date",min($"dt2") over Window.partitionBy($"user_id",$"tem"))
      .where($"count".>=(3) and $"count_carbon".>=(100))
      .show()





  }

}
